package com.lrd.rmqcluster.rmq5;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

@Component
@Order(1000)
public class MqConsumer implements ApplicationContextAware {

    private static final Logger log = LoggerFactory.getLogger(MqConsumer.class);

    private final List<DefaultMQPushConsumer> consumerList = new ArrayList<>();

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;
    @Value("${apache.rocketmq.test.topic}")
    private String topic;
    @Value("${apache.rocketmq.test.group}")
    private String group;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //初始化rocketmq消费者
        initTestConsumer(applicationContext);
    }

    private void initTestConsumer(ApplicationContext applicationContext) {
        try {
            log.info("=========================init-test-consumer=========================");
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
            consumer.setNamesrvAddr(namesrvAddr);
            consumer.setInstanceName("INSTANCE_NAME-" + UUID.randomUUID());
            consumer.setConsumeThreadMin(5);
            consumer.setConsumeThreadMax(5);
            consumer.setMaxReconsumeTimes(3);
            // 从上次开始的位置消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 集群消费模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            // 订阅主题
            consumer.subscribe(topic, "*");
            // 消息监听器
            consumer.registerMessageListener(applicationContext.getBean(MQMessageListener.class));
            consumer.start();
            consumerList.add(consumer);
            log.info("=========================test-consumer-start-success=========================");
        } catch (MQClientException e) {
            log.error("************************TestConsumer初始化启动失败\n************************", e);
        }
    }

    @PreDestroy
    public void stop() {
        for (DefaultMQPushConsumer consumer : consumerList) {
            if (consumer != null) {
                consumer.shutdown();
                log.info("MQ：stop\"" + consumer.getInstanceName() + "\"Consumer success! ");
            }
        }

    }
}
